package org.kth.dks.dks_marshal;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.log4j.Logger;
import org.kth.dks.DKSObject;
import org.kth.dks.DKSObjectTypes;
import org.kth.dks.dks_comm.DKSNetAddress;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;

/**
 * <p>Title: DKS</p>
 *
 * <p>Description: DKS Middleware</p>
 *
 * <p>Copyright: Copyright (c) 2005</p>
 *
 * <p>Company: KTH-IMIT/SICS</p>
 *
 * @author Ali Ghodsi (aligh@kth.se)
 * @version 1.0
 */

public class MarshalBinary extends MarshalInterface {

  private Logger log = Logger.getLogger(MarshalBinary.class);
  private DataInputStream rawIn;
  private DataOutputStream rawOut;

  private static final String BASEPACKAGE = "org.kth.dks.";

  public MarshalBinary(DKSMessage m) {
    super(m);
  }

  public byte[] flatten() {
    ByteArrayOutputStream as = new ByteArrayOutputStream();
    rawOut = new DataOutputStream(as);

    try {
      rawOut.writeUTF(msg.getName());

      rawOut.writeBoolean(msg.getSendRef() == null ? false : true);
      if (msg.getSendRef() != null) msg.marshaler.addDKSRef(msg.getSendRef(), "SrcRef");

      rawOut.writeBoolean(msg.getRecvRef() == null ? false : true);
      if (msg.getRecvRef() != null) msg.marshaler.addDKSRef(msg.getRecvRef(), "DstRef");

      msg.marshal();

    }
    catch (IOException ex) {
      log.error( "Error marshaling:\n" + ex);
    }

    byte[] ret = as.toByteArray();

    as = null;
    rawOut = null;
    return ret;
  }

  public void unflatten(DataInputStream raw) {
    rawIn = raw;
    try {

      final boolean anysrc = msg.marshaler.remBool();
      if (anysrc) msg.setSendRef(msg.marshaler.remDKSRef("SrcRef"));
      final boolean anydest = msg.marshaler.remBool();
      if (anydest) msg.setRecvRef(msg.marshaler.remDKSRef("DstRef"));

      msg.unmarshal();
    }
    catch (IOException ex) {
      rawIn = null;
      log.error( "Error unmarshaling:\n" + ex);
    }
    rawIn = null;
  }

  public static DKSMessage unflatten(byte[] raw) {
    try { // minimize use of this function, because it will be synchronized!
      ByteArrayInputStream bin = new ByteArrayInputStream(raw);
      DataInputStream din = new DataInputStream(bin);
      String name = din.readUTF();

      DKSMessage msg = null;
      if (DKSMessage.messageMap.get(name) != null) {
        Class messageClass = (Class) DKSMessage.messageMap.get(name);
        msg = ( (DKSMessage) messageClass.newInstance());
        ( (MarshalBinary) msg.marshaler).unflatten(din);
      }
      return msg;
    }
    catch (Exception ex) {
      ex.printStackTrace();
    }
    return null;
  }

  // ############# STANDARD HANDLERS ################

  public final void addLong(long val) throws IOException {
    addLong(val, null);
  }

  public final void addLong(long val, String na) throws IOException {
    rawOut.writeLong(val);
  }

  public final void addInt(int val) throws IOException {
    addInt(val, null);
  }

  public final void addInt(int val, String na) throws IOException {
    rawOut.writeInt(val);
  }

  public final void addString(String val) throws IOException {
    addString(val, null);
  }

  public final void addString(String val, String na) throws IOException {
    rawOut.writeUTF(val);
  }

  public final void addBool(boolean val) throws IOException {
    addBool(val, null);
  }

  public final void addBool(boolean val, String na) throws IOException {
    rawOut.writeBoolean(val);
  }

  public final void addDKSObject(DKSObject val) throws IOException {
    addDKSObject(val, null);
  }

  public final void addDKSObject(DKSObject val, String na) throws IOException {
    addString(val.getType().getName());
    rawOut.writeInt(val.getData().length);
    rawOut.write(val.getData());
  }

  public final void addDKSRef(DKSRef val) throws IOException {
    addDKSRef(val, null);
  }

  public final void addDKSRef(DKSRef val, String na) throws IOException {
    addDKSNetAddress(val.getDKSNetAddress(), null);
    addDKSOverlayAddress(val.getOverlayAddress());
    addLong(val.getNonce());
  }

  public final void addDKSNetAddress(DKSNetAddress val) throws IOException {
    addDKSNetAddress(val, null);
  }

  public final void addDKSNetAddress(DKSNetAddress val, String na) throws IOException {
    addString(val.getIP());
    addInt(val.getPort());
  }

  public final void addDKSOverlayAddress(DKSOverlayAddress val) throws IOException {
    addLong(val.getID());
    addLong(val.getGroupId());
    addLong(val.getGUID());
  }

  public final void addDKSRefArray(DKSRef[] val) throws IOException {
    addDKSRefArray(val, null);
  }

  public final void addDKSRefArray(DKSRef[] val, String na) throws IOException {
    addInt(val.length);
    for(int i=0; i<val.length; i++) {
      addDKSRef(val[i]);
    }
  }

  public final void addByteArray(byte[] m) throws IOException {
    addByteArray(m, "");
  }

  public final void addByteArray(byte[] m, String na) throws IOException {
    addInt(m.length);
    rawOut.write(m, 0, m.length);
  }

  // ######################## REMOVE OPERATIONS ##########################

  public final String remString() throws IOException {
    return remString(null);
  }

  public final String remString(String na) throws IOException {
    return rawIn.readUTF();
  }

  public final long remLong() throws IOException {
    return remLong(null);
  }

  public final long remLong(String na) throws IOException {
    return rawIn.readLong();
  }

  public final int remInt() throws IOException {
    return remInt(null);
  }

  public final int remInt(String na) throws IOException {
    return rawIn.readInt();
  }

  public final boolean remBool() throws IOException {
    return remBool(null);
  }

  public final boolean remBool(String na) throws IOException {
    return rawIn.readBoolean();
  }

  public final DKSObject remDKSObject() throws IOException {
    return remDKSObject(null);
  }

  public final DKSObject remDKSObject(String na) throws IOException {
    String type = remString();
    int len = remInt();
    byte[] obj = new byte[len];
    rawIn.read(obj, 0, len);
    return new DKSObject(DKSObjectTypes.valueOf(type), obj);
  }

  public final DKSRef remDKSRef() throws IOException {
    return remDKSRef(null);
  }

  public final DKSRef remDKSRef(String na) throws IOException {
    DKSNetAddress netadr     = remDKSNetAddress();
    DKSOverlayAddress ovradr = remDKSOverlayAddress();
    return DKSRef.valueOfByParts(ovradr, netadr, remLong());
  }

  public final DKSNetAddress remDKSNetAddress() throws IOException {
    return remDKSNetAddress(null);
  }

  public final DKSNetAddress remDKSNetAddress(String na) throws IOException {
      return new DKSNetAddress(remString(), remInt());
  }

  public final DKSOverlayAddress remDKSOverlayAddress() throws IOException {
    return new DKSOverlayAddress(remLong(), remLong(), remLong());
  }

  public final List remDKSRefArray() throws IOException {
    return remDKSRefArray(null);
  }

  public final List remDKSRefArray(String na) throws IOException {
    int len = remInt();
    List arr = new ArrayList(len+1);
    for (int i = 0; i < len; i++) {
      arr.add(remDKSRef());
    }
    return arr;
  }

  public final  byte[] remByteArray() throws IOException {
    return remByteArray("");
  }

  public final byte[] remByteArray(String na) throws IOException {
    int s = remInt();
    byte[] ret = new byte[s];
    rawIn.read(ret, 0, s);
    return ret;
  }

}
